package com.feidee.fd.sml.algorithm.component.output

import com.feidee.fd.sml.algorithm.component.{AbstractComponent, BasicParam}
import org.apache.spark.sql.{DataFrame, SaveMode, functions}

import scala.util.matching.Regex

/**
  * @Author: dongguosheng, songhaicheng
  * @Date: 2018/8/16 10:19
  */
case class HiveOutputParam(
                            override val input_pt: String,
                            override val output_pt: String,
                            override val hive_table: String,
                            override val flow_time: String,
                            partition: String
                          ) extends BasicParam {

  def this() = this(null, null, null, null, null)

  override def verify(): Unit = {
    super.verify()
    require(tool.isNotNull(input_pt), "param input_pt can't be null")
    require(tool.isNotNull(hive_table), "param hive_table can't be null")
  }
}


class HiveOutput extends AbstractComponent[HiveOutputParam] {

  override def outputData(data: DataFrame, param: HiveOutputParam): Unit = {
    if (tool.isNotNull(param.partition)) {
      logInfo(s"Using partition ${param.partition}")
      // 先解析时间分区
      val part = PartitionParamParser.datePartition(param.partition)
      require(part.nonEmpty && part.length == 2, "解析日期分区信息出错，暂只支持 a = {b} 形式（a -> 分区字段，b -> 时间格式）")
      // 补上分区值，按分区插入目标数据库
      data.withColumn(part.head, functions.lit(part.last))
        .write.mode(SaveMode.Append).partitionBy(part.head).saveAsTable(param.hive_table)
    } else {
      data.sqlContext.sql("DROP TABLE IF EXISTS " + param.hive_table)
      data.write.saveAsTable(param.hive_table)
    }
  }
}

object HiveOutput {
  val component = new HiveOutput()

  def apply(paramStr: String): Unit = {
    component(paramStr)
  }

  def main(args: Array[String]): Unit = {
    val component = HiveOutput
    component(args(0))
  }
}

object PartitionParamParser {
  // e.g. ymd = {yyyy-MM-dd}
  val FORMULA_REG: Regex = "\\s*(\\w*)\\s+=\\s+\\{(.*)\\}\\s*".r

  def datePartition(str: String): List[String] = {
    val isMatch = FORMULA_REG.findAllIn(str).matchData.hasNext
    if (isMatch) {
      FORMULA_REG.findAllIn(str).matchData.next.subgroups
    }
    else {
      Array.empty[String]
      List.empty
    }
  }

  def main(args: Array[String]): Unit = {
    val f = datePartition(" ymd = {yyyy-MM-dd}")
    val format = new java.text.SimpleDateFormat(f(1))
    println(format.format(new java.util.Date()))
  }

}
